package com.example.reactor;

import org.reactivestreams.Subscription;
import reactor.core.publisher.BaseSubscriber;
import reactor.core.publisher.Flux;
import reactor.core.publisher.SignalType;

public class FluxDemo4 {

    public static void main(String[] args) {

        Flux<String> flux = Flux.range(1, 9)
                .map(integer -> {
                    if (integer == 5) {
                        int i = 1 / 0;
                    }
                    return "数据：" + integer;
                })
                .onErrorComplete();

//        flux.subscribe(
//                /**
//                 * OnNext
//                 */
//                v -> System.out.println(v + "-》嘿嘿"),
//                /**
//                 * OnError
//                 */
//                e -> System.out.println("出错了：" + e),
//                /**
//                 * OnComplete
//                 */
//                () -> System.out.println("完成")
//        );

        /**
         * 发布者流生命周期能过传递给订阅者
         */

        flux.subscribe(new BaseSubscriber<String>() {
            /**
             * 生命周期钩子：订阅关系绑定时触发
             * @param subscription the subscription to optionally process
             */
            @Override
            protected void hookOnSubscribe(Subscription subscription) {
                System.out.println("订阅关系" + subscription);
                request(1);
            }

            /**
             * 生命周期钩子：接收数据时触发
             * @param value the emitted value to process
             */
            @Override
            protected void hookOnNext(String value) {
                if (value.equals("数据：3")) {
                    cancel();
                } else {
                    System.out.println(value);
                    request(1);
                }
            }

            /**
             * 生命周期钩子：完成时触发
             */
            @Override
            protected void hookOnComplete() {
                System.out.println("完成");
            }

            /**
             * 生命周期钩子：异常时触发
             * @param throwable the exception to process
             */
            @Override
            protected void hookOnError(Throwable throwable) {
                System.out.println("异常：" + throwable);
            }

            /**
             * 生命周期钩子：取消时触发
             */
            @Override
            protected void hookOnCancel() {
                System.out.println("取消");
            }

            /**
             * 生命周期钩子：完成时触发
             * @param type the signal type
             */
            @Override
            protected void hookFinally(SignalType type) {
                System.out.println("最终：" + type);
            }
        });

    }

}
